%%%-------------------------------------------------------------------
%%% @author wukai
%%% @copyright (C) 2020, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 29. 8月 2020 9:34 上午
%%%-------------------------------------------------------------------
-module(himq_mqtt_codec).
-author("wukai").
-include("himq.hrl").
-include_lib("stdlib/include/qlc.hrl").
-define(FRAME_CONN, 1).
-define(FRAME_CONN_ACK, 2).
-define(FRAME_PUB, 3).
-define(FRAME_PUB_ACK, 4).
-define(FRAME_PUB_REC, 5).
-define(FRAME_PUB_REL, 6).
-define(FRAME_PUB_COMP, 7).
-define(FRAME_SUB, 8).
-define(FRAME_SUB_ACK, 9).
-define(FRAME_UNSUB, 10).
-define(FRAME_UNSUB_ACK, 11).
-define(FRAME_PING, 12).
-define(FRAME_PING_ACK, 13).
-define(FRAME_DISCONNECT, 14).
-define(WAIT_HEAD, wait_head).
-define(WAIT_LEFT_LEN, wait_left_len).
-define(WAIT_LEFT_BYTES, wait_left_bytes).

%%0	0x00连接已接受	连接已被服务端接受
%%1	0x01连接已拒绝，不支持的协议版本	服务端不支持客户端请求的MQTT协议级别
%%2	0x02连接已拒绝，不合格的客户端标识符	客户端标识符是正确的UTF-8编码，但服务端不允许使用
%%3	0x03连接已拒绝，服务端不可用	网络连接已建立，但MQTT服务不可用
%%4	0x04连接已拒绝，无效的用户名或密码	用户名或密码的数据格式无效
%%5	0x05连接已拒绝，未授权	客户端未被授权连接到此服务器

-define(CONNECT_CODE_OK, 0).
-define(CONNECT_CODE_BAD_PROTOCOL_VERSION, 1).
-define(CONNECT_CODE_BAD_ID_RULE, 2).
-define(CONNECT_CODE_BAD_SEVER, 3).
-define(CONNECT_CODE_BAD_AUTH_USER, 4).
-define(CONNECT_CODE_BAD_AUTH_ID, 5).


-record(mqtt, {wait_content, wait_length, wf, will, pv, clean, packet_id = 1, auth = false, remain_len = [], router :: pid(), frame = <<>>}).
%% API
-export([decode/1, p_init/0, timeout/1, closed/1, message/2, session_remark/1]).
p_init() -> #mqtt{wait_content = ?WAIT_HEAD}.

decode(State) when State#himq_conn.p_state#mqtt.wait_content == ?WAIT_HEAD -> head(State#himq_conn.buffer, State);
decode(State) when State#himq_conn.p_state#mqtt.wait_content == ?WAIT_LEFT_LEN -> remain_len(State);
decode(State) when State#himq_conn.p_state#mqtt.wait_content == ?WAIT_LEFT_BYTES ->
  remain(State, State#mqtt.wait_length).

head(<<?FRAME_CONN:4, _:4, _/binary>>, State) ->
  <<Head:8, Left/binary>> = State#himq_conn.buffer,
  NMqtt = State#himq_conn.p_state#mqtt{wait_content = ?WAIT_LEFT_LEN, frame = <<Head:8>>},
  NState = State#himq_conn{buffer = Left, p_state = NMqtt},
  remain_len(NState);

head(<<?FRAME_PUB:4, _:4, Left/binary>>, State) ->
  <<Head:8, Left/binary>> = State#himq_conn.buffer,
  NMqtt = State#himq_conn.p_state#mqtt{wait_content = ?WAIT_LEFT_LEN, frame = <<Head:8>>},
  NState = State#himq_conn{buffer = Left, p_state = NMqtt},
  remain_len(NState);
head(<<?FRAME_PUB_ACK:4, _:4, Left/binary>>, State) ->
  <<Head:8, Left/binary>> = State#himq_conn.buffer,
  NMqtt = State#himq_conn.p_state#mqtt{wait_content = ?WAIT_LEFT_LEN, frame = <<Head:8>>},
  NState = State#himq_conn{buffer = Left, p_state = NMqtt},
  remain_len(NState);

head(<<?FRAME_PUB_REC:4, _:4, Left/binary>>, State) ->
  <<Head:8, Left/binary>> = State#himq_conn.buffer,
  NMqtt = State#himq_conn.p_state#mqtt{wait_content = ?WAIT_LEFT_LEN, frame = <<Head:8>>},
  NState = State#himq_conn{buffer = Left, p_state = NMqtt},
  remain_len(NState);

head(<<?FRAME_PUB_REL:4, _:4, Left/binary>>, State) ->
  <<Head:8, Left/binary>> = State#himq_conn.buffer,
  NMqtt = State#himq_conn.p_state#mqtt{wait_content = ?WAIT_LEFT_LEN, frame = <<Head:8>>},
  NState = State#himq_conn{buffer = Left, p_state = NMqtt},
  remain_len(NState);

head(<<?FRAME_PUB_COMP:4, _:4, Left/binary>>, State) ->
  <<Head:8, Left/binary>> = State#himq_conn.buffer,
  NMqtt = State#himq_conn.p_state#mqtt{wait_content = ?WAIT_LEFT_LEN, frame = <<Head:8>>},
  NState = State#himq_conn{buffer = Left, p_state = NMqtt},
  remain_len(NState);

head(<<?FRAME_SUB:4, _:4, Left/binary>>, State) ->
  <<Head:8, Left/binary>> = State#himq_conn.buffer,
  NMqtt = State#himq_conn.p_state#mqtt{wait_content = ?WAIT_LEFT_LEN, frame = <<Head:8>>},
  NState = State#himq_conn{buffer = Left, p_state = NMqtt},
  remain_len(NState);

head(<<?FRAME_SUB_ACK:4, _:4, Left/binary>>, State) ->
  <<Head:8, Left/binary>> = State#himq_conn.buffer,
  NMqtt = State#himq_conn.p_state#mqtt{wait_content = ?WAIT_LEFT_LEN, frame = <<Head:8>>},
  NState = State#himq_conn{buffer = Left, p_state = NMqtt},
  remain_len(NState);

head(<<?FRAME_UNSUB:4, _:4, Left/binary>>, State) ->
  <<Head:8, Left/binary>> = State#himq_conn.buffer,
  NMqtt = State#himq_conn.p_state#mqtt{wait_content = ?WAIT_LEFT_LEN, frame = <<Head:8>>},
  NState = State#himq_conn{buffer = Left, p_state = NMqtt},
  remain_len(NState);

head(<<?FRAME_PING:4, _:4, Left/binary>>, State) ->
  <<Head:8, Left/binary>> = State#himq_conn.buffer,
  NMqtt = State#himq_conn.p_state#mqtt{wait_content = ?WAIT_LEFT_LEN, frame = <<Head:8>>},
  NState = State#himq_conn{buffer = Left, p_state = NMqtt},
  remain_len(NState);


head(<<?FRAME_DISCONNECT:4, _:4, Left/binary>>, State) ->
  <<Head:8, Left/binary>> = State#himq_conn.buffer,
  NMqtt = State#himq_conn.p_state#mqtt{wait_content = ?WAIT_LEFT_LEN, frame = <<Head:8>>},
  NState = State#himq_conn{buffer = Left, p_state = NMqtt},
  remain_len(NState);


head(<<_Head:8, Left/binary>>, State) ->
  NState = State#himq_conn{buffer = Left},
  head(Left, NState);
head(<<>>, State) ->
  State#himq_conn{buffer = <<>>}.


%%不够1个字节继续等待接收
remain_len(State) when byte_size(State#himq_conn.buffer) < 1 ->
  NPState = State#himq_conn.p_state#mqtt{wait_content = ?WAIT_LEFT_LEN},
  State#himq_conn{p_state = NPState};

remain_len(State) ->
  <<First:1, _:7, Left/binary>> = State#himq_conn.buffer,
  <<Len:8, _/binary>> = State#himq_conn.buffer,
  remain_len(First, Left, State, Len).

remain_len(1, Left, State, Len) ->
  Lens = State#himq_conn.p_state#mqtt.remain_len ++ [Len],
  NPState = State#himq_conn.p_state#mqtt{remain_len = Lens},
  remain_len(State#himq_conn{buffer = Left, p_state = NPState});

remain_len(0, Left, State, Len) ->
  Len_bytes = State#himq_conn.p_state#mqtt.remain_len ++ [Len],
  Remain_Len = bytes_to_len(Len_bytes),
  remain(State#himq_conn{buffer = Left}, Remain_Len).

remain(State, Len) when byte_size(State#himq_conn.buffer) < Len ->
  NPState = State#himq_conn.p_state#mqtt{wait_content = ?WAIT_LEFT_BYTES, wait_length = Len},
  State#himq_conn{p_state = NPState};

remain(State, Len) ->
  <<Play:Len/binary, Left/binary>> = State#himq_conn.buffer,
  Head = State#himq_conn.p_state#mqtt.frame,
  NPState = State#himq_conn.p_state#mqtt{remain_len = [], wait_content = ?WAIT_HEAD},
  NewState = frame(<<Head/binary, Play/binary>>, State#himq_conn{buffer = Left, p_state = NPState}),
  case NewState of
    {error, Reason} ->
      lager:info("frame error ~p", [Reason]),
      exit(normal);
    _ -> decode(NewState)
  end.

frame(<<?FRAME_CONN:4, _:4, _:?SHORT, "MQTT", 4:?BYTE, _/binary>>, State) when State#himq_conn.p_state#mqtt.auth == true ->
  State;
frame(<<?FRAME_CONN:4, _:4, _:?SHORT, "MQTT", 4:?BYTE, 1:1, 1:1, WR:1, WQ:2, 0:1, CS:1, 0:1, KA:?SHORT, IDL:16, CID:IDL/binary, PL/binary>>,
    State) ->
  lager:info("WF:~p WR:~p,WQ:~p,CS:~p,KA:~p", [0, WR, WQ, CS, KA]),
  ClientId = binary_to_list(CID),
  <<UserLen:?SHORT, UserName:UserLen/binary, PassLen:16, Password:PassLen/binary>> = PL,
  AuthWay = himq_eenv:get(himq_app, ?SETTING_BASE_KEY, #{auth_way=><<"user_pass">>}),
  lager:info("auth_way ~p", [AuthWay]),
  case auth(AuthWay, UserName, Password, ClientId) of
    true ->
      himq_mqtt:send(State, encode(?FRAME_CONN_ACK, ?CONNECT_CODE_OK)),
      %%1.5倍的时间超时
      Router = himq_mqtt_router:check_router(),
      PState = State#himq_conn.p_state#mqtt{wf = 0, pv = 4, router = Router, clean = CS, auth = true},
      himq_session:add(?MODULE, mqtt, 4, State#himq_conn.socket, ClientId, State#himq_conn.type, #{clean=>CS, wf=>0,username=>UserName}, self(), self()),
      NState = send_undelivered(ClientId, State),
      NState#himq_conn{client_id = ClientId, receive_timeout = KA * 100 * 15, p_state = PState};
    false ->
      himq_mqtt:send(State, encode(?FRAME_CONN_ACK, ?CONNECT_CODE_BAD_AUTH_USER)),
      lager:info("auth error"),
      {error, "auth error"}
  end;

frame(<<?FRAME_CONN:4, _:4, _:?SHORT, "MQTT", 4:?BYTE, 1:1, 1:1, WR:1, WQ:2, 1:1, CS:1, 0:1, KA:?SHORT, IDL:16, CID:IDL/binary, PL/binary>>,
    State) ->
  lager:info("WF:~p WR:~p,WQ:~p,CS:~p,KA:~p", [0, WR, WQ, CS, KA]),
  ClientId = binary_to_list(CID),
  <<TopLen:16, Topic:TopLen/binary, MsgLen:16, Msg:MsgLen/binary, Left/binary>> = PL,
  <<UserLen:?SHORT, UserName:UserLen/binary, PassLen:16, Password:PassLen/binary>> = Left,
  %%获取认证方式
  AuthWay = himq_eenv:get(himq_app, ?SETTING_BASE_KEY, #{auth_way=><<"user_pass">>}),
  lager:info("auth_way ~p", [AuthWay]),
  case auth(AuthWay, UserName, Password, ClientId) of
    true ->
      himq_mqtt:send(State, encode(?FRAME_CONN_ACK, ?CONNECT_CODE_OK)),
      %%1.5倍的时间超时
      Router = himq_mqtt_router:check_router(),
      Will = #{wr=>WR, wqos=>WQ, wtopic=>Topic, wmsg=>Msg},

      PState = State#himq_conn.p_state#mqtt{wf = 1, pv = 4, router = Router, will =Will, clean = CS, auth = true},
      himq_session:add(?MODULE, mqtt, 4, State#himq_conn.socket, ClientId, State#himq_conn.type,
        #{clean=>CS, wf=>1, will=>Will,username=>UserName}, self(), self()),
      NState = send_undelivered(ClientId, State),
      NState#himq_conn{client_id = ClientId, receive_timeout = KA * 100 * 15, p_state = PState};
    false ->
      himq_mqtt:send(State, encode(?FRAME_CONN_ACK, ?CONNECT_CODE_BAD_AUTH_USER)),
      lager:info("auth error"),
      {error, "auth error"}
  end;

%%QOS=1
frame(<<?FRAME_PUB:4, Dup:1, 1:2, Retain:1, Binary/binary>>, State) ->
  <<TLen:16, Topic:TLen/binary, Id:16, Msg/binary>> = Binary,
  save_retain(State#himq_conn.client_id, {Dup, 1, Retain, Topic, Msg}),
  NState = in_router(State, {Dup, 0, Retain, Topic, Msg}),
  himq_mqtt:send(NState, encode(?FRAME_PUB_ACK, Id)), NState;

%%QOS==2
frame(<<?FRAME_PUB:4, Dup:1, 2:2, Retain:1, Binary/binary>>, State) ->
  <<TLen:16, Topic:TLen/binary, PacketId:16, Msg/binary>> = Binary,
  Publish = {Dup, 2, Retain, Topic, Msg},
  put({msg, PacketId}, {Publish, wait_rel, received}),
  himq_mqtt:send(State, encode(?FRAME_PUB_REC, PacketId)), State;

%%QOS=0
frame(<<?FRAME_PUB:4, Dup:1, 0:2, Retain:1, Binary/binary>>, State) ->
  <<TLen:16, Topic:TLen/binary, Msg/binary>> = Binary,
  save_retain(State#himq_conn.client_id, {Dup, 0, Retain, Topic, Msg}),
  in_router(State, {Dup, 0, Retain, Topic, Msg});

frame(<<?FRAME_SUB:4, _:4, Binary/binary>>, State) ->
  <<PacketID:16, Filters/binary>> = Binary,
  List = subscribe(Filters, []),
  himq_mqtt_router:add_router(List, PacketID, State#himq_conn.client_id, self(), State#himq_conn.p_state#mqtt.clean),
  State;

frame(<<?FRAME_UNSUB:4, _:4, Binary/binary>>, State) ->
  <<PacketID:16, Filters/binary>> = Binary,
  List = cancel_subscribe(Filters, []),
  himq_mqtt_router:del_router(List, State#himq_conn.client_id, self()),
  himq_mqtt:send(State, encode(?FRAME_UNSUB_ACK, PacketID)),
  State;

frame(<<?FRAME_PUB_ACK:4, _:4, Binary/binary>>, State) ->
  <<PacketID:16, _/binary>> = Binary,
  Key = {msg, PacketID},
  case get(Key) of
    %%api push的 qos=1的消息
    {Push, wait_ack, push} ->
      erase(Key),
      himq_mnesia:write(Push#himq_push{end_time = himq:timestamp(), status = acked});
    %% 来自router的消息 删除消息
    {_Publish, wait_ack, router} ->
      erase(Key);
    %%离线的消息
    {Undelivered, wait_ack, undelivered} ->
      himq_mnesia:delete(?TABLE_UNDELIVERED, Undelivered#himq_undelivered.uuid);
    _ -> ok
  end,
  State;

frame(<<?FRAME_PUB_REC:4, _:4, Binary/binary>>, State) ->
  <<PacketID:16, _/binary>> = Binary, {?FRAME_PUB_REC, PacketID},
  Key = {msg, PacketID},
  case get(Key) of
    %%api push的 qos=1的消息
    {Push, wait_rec, push} ->
      put(Key, {Push, wait_comp, push});
    %% 来自router的消息 删除消息
    {Publish, wait_rec, router} ->
      put(Key, {Publish, wait_comp, router});
    %%离线的消息
    {Undelivered, wait_rec, undelivered} ->
      himq_mnesia:write(Undelivered#himq_undelivered{step = wait_comp}),
      put(Key, {Undelivered, wait_comp, undelivered});
    _ -> ok
  end,
  himq_mqtt:send(State, encode(?FRAME_PUB_REL, PacketID)),
  State;

frame(<<?FRAME_PUB_REL:4, _:4, Binary/binary>>, State) ->
  <<PacketID:16, _/binary>> = Binary, {?FRAME_PUB_REL, PacketID},
  Key = {msg, PacketID},
  NState = case get(Key) of
             %% 来自router的消息 删除消息
             {Publish, wait_rel, received} ->
               erase(Key),
               save_retain(State#himq_conn.client_id, Publish),
               in_router(State, Publish);
             _ ->
               V = get(Key),
               lager:info("pub_rel msg: key:~p,Publish ~p", [Key, V]),
               State
           end,
  himq_mqtt:send(State, encode(?FRAME_PUB_COMP, PacketID)),
  NState;

frame(<<?FRAME_PUB_COMP:4, _:4, Binary/binary>>, State) ->
  <<PacketID:16, _/binary>> = Binary, {?FRAME_PUB_COMP, PacketID},
  Key = {msg, PacketID},
  case get(Key) of
    %%api push的 qos=1的消息
    {Push, wait_comp, push} ->
      erase(Key),
      himq_mnesia:write(Push#himq_push{end_time = himq:timestamp(), status = pub_comp});
    %% 来自router的消息 删除消息
    {_Publish, wait_comp, router} ->
      erase(Key);
    %%离线消息
    {Undelivered, wait_comp, undelivered} ->
      erase(Key),
      lager:error("receive pub comp PacketId ~p", [PacketID]),
      himq_mnesia:delete(?TABLE_UNDELIVERED, Undelivered#himq_undelivered.uuid);
    _ -> ok
  end,
  State;


frame(<<?FRAME_PING:4, 0:4, _/binary>>, State) ->
  lager:info("received ping from ~p", [State#himq_conn.client_id]),
  himq_mqtt:send(State, encode(?FRAME_PING_ACK, 0)),
  %%进程休眠
  himq_mqtt:hibernate(State);

frame(<<?FRAME_DISCONNECT:4, 0:4, _/binary>>, State) ->
  save_wait_ack_pub(State#himq_conn.client_id),
  State.


%%{id,uuid,retain,dup,topic,qos,packet,from,step,remark,created_at,updated_at}).
save_wait_ack_pub(Client) ->
  List = get(),
  lists:foreach(
    fun(E) ->
      case E of
        {{msg, PacketId}, {From, {Dup, Qos, 0, Topic, Packet}, Step, router}} ->
          Uuid = packet_uuid(PacketId),
          Item = #himq_undelivered{id = Client, from = From, uuid = Uuid, topic = Topic, dup = Dup, qos = Qos, step = Step, packet = Packet, created_at = himq:timestamp()},
          Result = himq_mnesia:write(Item),
          lager:info("save unack message result:", [Result])
      end
    end,
    List).

remain_len_bytes(Remain) -> len_to_bytes(byte_size(Remain)).

encode(?FRAME_CONN_ACK, Code) ->
  FH = fh({?FRAME_CONN_ACK}),
  VH = <<1:8>>,
  Remain = <<VH/binary, Code:?BYTE>>,
  frame_binary(FH, Remain);

%% encode subAck
encode(?FRAME_SUB_ACK, Args) ->
  {ID, Bin} = Args,
  FH = fh({?FRAME_SUB_ACK}),
  Remain = <<ID:?SHORT, Bin/binary>>,
  frame_binary(FH, Remain);

encode(?FRAME_UNSUB_ACK, PacketID) ->
  FH = fh({?FRAME_SUB_ACK}),
  Remain = <<PacketID:?SHORT>>,
  frame_binary(FH, Remain);

%%pub
encode(?FRAME_PUB, {{Dup, Qos, Retain, Topic, Packet}, PacketID}) ->
  FH = fh({?FRAME_PUB, Dup, Qos, Retain}),
  TLen = byte_size(Topic),
  Remain = <<TLen:?SHORT, Topic/binary, PacketID:?SHORT, Packet/binary>>,
  frame_binary(FH, Remain);

encode(?FRAME_PUB, {Dup, Qos, Retain, Topic, Packet}) ->
  FH = fh({?FRAME_PUB, Dup, Qos, Retain}),
  TLen = byte_size(Topic),
  Remain = <<TLen:?SHORT, Topic/binary, Packet/binary>>,
  frame_binary(FH, Remain);

%%pub_ack
encode(?FRAME_PUB_ACK, PacketID) ->
  FH = fh({?FRAME_PUB_ACK}),
  Remain = <<PacketID:?SHORT>>,
  frame_binary(FH, Remain);
%%pub_ack
encode(?FRAME_PUB_REL, PacketID) ->
  FH = fh({?FRAME_PUB_REL}),
  Remain = <<PacketID:?SHORT>>,
  frame_binary(FH, Remain);
encode(?FRAME_PUB_REC, PacketID) ->
  FH = fh({?FRAME_PUB_REC}),
  Remain = <<PacketID:?SHORT>>,
  frame_binary(FH, Remain);

encode(?FRAME_PUB_COMP, PacketID) ->
  FH = fh({?FRAME_PUB_COMP}),
  Remain = <<PacketID:?SHORT>>,
  frame_binary(FH, Remain);

encode(?FRAME_PING, _) ->
  FH = fh({?FRAME_PING}),
  <<FH/binary, 0:?BYTE>>;

encode(?FRAME_PING_ACK, _) ->
  FH = fh({?FRAME_PING_ACK}),
  <<FH/binary, 0:?BYTE>>.

frame_binary(Head, Remain) ->
  RLen = remain_len_bytes(Remain),
  <<Head/binary, RLen/binary, Remain/binary>>.

bytes_to_len(List) -> bytes_to_len(List, 0, 1).
bytes_to_len([H | T], Value, Index) ->
  Value1 = len_multi(H, Index) + Value, NIndex = Index + 1,
  bytes_to_len(T, Value1, NIndex);
bytes_to_len([], Value, _) -> Value.

len_multi(Byte, 1) ->
  Byte band 127;
len_multi(Byte, 2) ->
  (Byte band 127) * 128;
len_multi(Byte, 3) ->
  (Byte band 127) * 128 * 128;
len_multi(Byte, 4) ->
  (Byte band 127) * 128 * 128 * 128.


%%clean msg retain ==1 msg==<<>>
save_retain(_Client, {_Dup, _Qos, 1, Topic, <<>>}) ->
  case himq_mnesia:delete(?TABLE_RETAIN, Topic) of
    {atomic, ok} ->
      lager:info("delete  retain mesage ok topic ~p ", [Topic]);
    _ ->
      lager:info("delete  retain mesage error topic ~p ", [Topic])
  end;

save_retain(Client, {_Dup, Qos, 1, Topic, Packet}) ->
  F = fun() ->
    case mnesia:read({?TABLE_RETAIN, Topic}) of
      [] ->
        Item = #himq_retain{qos = Qos, topic = Topic, created_at = himq:timestamp(), updated_at = himq:timestamp(), id = Client},
        himq_mnesia:write(Item);
      [Retain] ->
        himq_mnesia:write(Retain#himq_retain{updated_at = himq:timestamp(), packet = Packet, qos = Qos})
    end
      end,
  mnesia:transaction(F);

save_retain(_, {_, _, 0, _, _}) -> ok.

-spec wrap_publish(integer(), integer(), integer(), binary(), binary()) -> tuple().
wrap_publish(Dup, Qos, Retain, Topic, Packet) -> {Dup, Qos, Retain, Topic, Packet}.
-spec subscribe(binary(), any()) -> any().
subscribe(<<>>, List) -> List;
subscribe(Binary, List) ->
  <<TLen:16, Topic:TLen/binary, Qos:8, Rest/binary>> = Binary,
  %%okmq_util:info("sub filter ~p qos ~p",[Topic,Qos]),
  subscribe(Rest, List ++ [{Topic, Qos}]).

cancel_subscribe(<<>>, List) -> List;
cancel_subscribe(Binary, List) ->
  <<TLen:?SHORT, Topic:TLen/binary, Left/binary>> = Binary,
  cancel_subscribe(Left, List ++ [Topic]).

fh({?FRAME_CONN}) -> <<?FRAME_CONN:4, 0:4>>;
fh({?FRAME_CONN_ACK}) -> <<?FRAME_CONN_ACK:4, 0:4>>;
fh({?FRAME_PING_ACK}) -> <<?FRAME_PING_ACK:4, 0:4>>;
fh({?FRAME_PING}) -> <<?FRAME_PING:4, 0:4>>;
fh({?FRAME_DISCONNECT}) -> <<?FRAME_DISCONNECT:4, 0:4>>;
fh({?FRAME_PUB_ACK}) -> <<?FRAME_PUB_ACK:4, 0:4>>;
fh({?FRAME_PUB_REL}) -> <<?FRAME_PUB_REL:4, 0:4>>;
fh({?FRAME_PUB_REC}) -> <<?FRAME_PUB_REC:4, 0:4>>;
fh({?FRAME_PUB_COMP}) -> <<?FRAME_PUB_COMP:4, 0:4>>;
fh({?FRAME_SUB_ACK}) -> <<?FRAME_SUB_ACK:4, 0:4>>;
fh({?FRAME_PUB, Dup, Qos, Retain}) -> <<?FRAME_PUB:4, Dup:1, Qos:2, Retain:1>>.

to_binary(Source) when is_list(Source) -> list_to_binary(Source);
to_binary(Source) when is_atom(Source) -> atom_to_binary(Source, utf8);
to_binary(Source) when is_integer(Source) -> integer_to_binary(Source);
to_binary(Source) when is_binary(Source) -> Source.

%%剩余长度转为字节数组
len_to_bytes(Value) when Value == 0 -> <<0:8>>;
len_to_bytes(Value) -> len_to_bytes(Value, <<>>).

len_to_bytes(Value, Bin) when Value == 0 -> Bin;
len_to_bytes(Value, Bin) ->
  Rem = Value rem 128,
  DIV = Value div 128,
  if DIV > 0 ->
    Rem1 = Rem bor 128,
    len_to_bytes(DIV, <<Bin/binary, Rem1:8>>);
    true -> <<Bin/binary, Rem:8>>
  end.

auth(#{auth_way:=<<"user_pass">>}, Username, Password, _Id) ->
  check_user_pass(Username, Password);
auth(#{auth_way:=<<"id">>}, _Username, _Password, Id) ->
  himq_mnesia:record_exist(?TABLE_DEVICE, Id);
auth(#{auth_way:=<<"user_pass_id">>}, Username, Password, Id) ->
  himq_mnesia:record_exist(?TABLE_DEVICE, Id)
    and
    check_user_pass(Username, Password);

auth(_AuthWay, _Username, _Password, _Id) -> false.

check_user_pass(Username, Password) ->
  L = himq_eenv:get(himq_app, users, []),
  lists:member({Username, Password}, L).

timeout(State) ->
%%删除订阅者信息
  delete_router(State#himq_conn.client_id).


%%删除订阅
delete_router(Client) ->
  List = lists:foldl(fun(E, Acc) ->
    case E of
      {Key, sub_filter} -> Acc ++ [Key];
      {Key, _} -> Acc
    end end, [], get()),
  himq_mqtt_router:del_router(List, Client, self()).

closed(State) ->
  delete_router(State#himq_conn.client_id), State.

%%外部消息
message({sub_ack, PacketId, Bin, SubList}, State) ->
  lists:foreach(fun({T, _Qos}) -> put(T, sub_filter) end, SubList),
  himq_mqtt:send(State, encode(?FRAME_SUB_ACK, {PacketId, Bin})),
  send_retain(SubList, State);

message({pub, {_From, {Dup, 0, Retain, Topic, Packet}}}, State) ->
  himq_mqtt:send(State, encode(?FRAME_PUB, {Dup, 0, Retain, Topic, Packet})),
  State;

message({pub, {_From, {Dup, Qos, Retain, Topic, Packet}} = P}, State) ->
  %%生成packetId
  PacketId = packet_id(State#himq_conn.p_state#mqtt.packet_id),
  case Qos of
    1 -> put({msg, PacketId}, {P, wait_ack, router});
    2 -> put({msg, PacketId}, {P, wait_rec, router})
  end,
  himq_mqtt:send(State, encode(?FRAME_PUB, {{Dup, Qos, Retain, Topic, Packet}, PacketId})),
  PState = State#himq_conn.p_state#mqtt{packet_id = PacketId + 1},
  State#himq_conn{p_state = PState};

message({push, Push}, State) ->
  PacketId = packet_id(State#himq_conn.p_state#mqtt.packet_id),
  Dup = 0,
  %%#{push_qos:=Qos} = himq_eenv:get(himq_app, ?SETTING_PUSH_KEY, #{push_qos=><<"1">>}),
  %%Qos1 = binary_to_integer(Qos),
  %%NewPush = Push#himq_push{qos = Qos1},
  case Push#himq_push.qos of
    1 -> put({msg, PacketId}, {Push, wait_ack, push});
    2 -> put({msg, PacketId}, {Push, wait_rec, push});
    _ ->
      ok
  end,

  NewP = {Dup, Push#himq_push.qos, Push#himq_push.retain, Push#himq_push.topic, Push#himq_push.packet},
  himq_mqtt:send(State, encode(?FRAME_PUB, {NewP, PacketId})),
  PState = State#himq_conn.p_state#mqtt{packet_id = PacketId + 1},
  State#himq_conn{p_state = PState};

message(Msg, State) ->
  lager:error("received extra message hand to handle ~p", [Msg]),
  State.

send_undelivered(Client, State) ->
  %%Fun = fun() -> mnesia:read({?TABLE_UNDELIVERED, Client}) end,
  Fun = fun() ->
    QH = qlc:q([Item || Item <- mnesia:table(?TABLE_UNDELIVERED), Item#himq_undelivered.id == Client]),
    QC = qlc:cursor(QH),
    qlc:next_answers(QC, all_remaining)
        end,
  Ret = case mnesia:transaction(Fun) of
          {atomic, []} ->
            lager:info("undelivered size:0"),
            ok;
          {atomic, List} ->
            lager:error("undelivered size:~p", [length(List)]),
            PacketId = packet_id(State#himq_conn.p_state#mqtt.packet_id),
            lists:foldl(fun(E, NextPacketId) ->
              lager:info("NextPacketId ~p", [NextPacketId]),
              Publish = {E#himq_undelivered.dup,
                E#himq_undelivered.qos,
                E#himq_undelivered.retain,
                E#himq_undelivered.topic,
                E#himq_undelivered.packet
              },

              case E#himq_undelivered.qos of
                1 -> put({msg, NextPacketId}, {E, wait_ack, undelivered});
                2 -> put({msg, NextPacketId}, {E, wait_rec, undelivered});
                _ -> ok
              end,
              himq_mqtt:send(State, encode(?FRAME_PUB, {Publish, NextPacketId})),
              packet_id(NextPacketId + 1)
                        end, PacketId, List);
          Err ->
            lager:info("send undelivered ~p", [Err]),
            ok
        end,

  case Ret of
    ok -> State;
    Id when is_integer(Id) ->
      PState = State#himq_conn.p_state#mqtt{packet_id = Ret + 1},
      State#himq_conn{p_state = PState}
  end.

send_retain(SubList, State) ->
  %%查询retain消息
  Fun = fun() -> lists:foldl(fun({T, Qos}, AccIn) ->
    case mnesia:read({?TABLE_RETAIN, T}) of
      [] -> AccIn;
      List -> AccIn ++
      lists:map(fun(E) -> {Qos, E} end, List)
    end
                             end, [], SubList) end,

  Ret = case mnesia:transaction(Fun) of
          {atomic, []} -> ok;
          {atomic, RetainList} ->
            lager:info("retain list ~p", [RetainList]),
            PacketId = packet_id(State#himq_conn.p_state#mqtt.packet_id),
            lists:foldl(fun({NewQos, Retain}, NextPacketId) ->
              PubRetain = {0, NewQos, 1, Retain#himq_retain.topic, Retain#himq_retain.packet},
              himq_mqtt:send(State, encode(?FRAME_PUB, {PubRetain, NextPacketId})),

              packet_id(NextPacketId + 1)
                        end, PacketId, RetainList);
          Err ->
            lager:info("send retain message error ~p", [Err]), ok
        end,

  case Ret of
    ok -> State;
    Id when is_integer(Id) ->
      PState = State#himq_conn.p_state#mqtt{packet_id = Ret + 1},
      State#himq_conn{p_state = PState}
  end.
packet_id(Id) when Id > 60000 -> 1;
packet_id(Id) -> Id.

packet_uuid(PacketID) ->
  List_Node = atom_to_list(node()),
  PID = integer_to_list(PacketID),
  himq:md5(lists:concat([[PacketID] ++ List_Node ++ PID ++ erlang:tuple_to_list(erlang:timestamp()) ++ erlang:pid_to_list(self())])).

in_router(State, {Dup, Qos, Retain, Topic, Msg}) ->
  RouterPid = State#himq_conn.p_state#mqtt.router,
  case is_process_alive(RouterPid) of
    true ->
      himq_mqtt_router:dispatch(State#himq_conn.p_state#mqtt.router, State#himq_conn.client_id, {Dup, Qos, Retain, Topic, Msg}),
      State;
    false ->
      Pid = himq_mqtt_router:check_router(),
      himq_mqtt_router:dispatch(State#himq_conn.p_state#mqtt.router, State#himq_conn.client_id, {Dup, Qos, Retain, Topic, Msg}),
      PState = State#himq_conn.p_state#mqtt{router = Pid},
      State#himq_conn{p_state = PState}
  end.

session_remark(Remark) -> Remark.
